package com.at.bigdata.spark.streaming

import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

import java.util.Properties
import scala.collection.mutable.ListBuffer
import scala.util.Random

/**
 *
 * @author cdhuangchao3
 * @date 2023/5/29 9:24 PM
 */
object SparkStreaming10_MockData {
  private val switch = false

  def main(args: Array[String]): Unit = {
    // 生成模拟数据
    // 格式： timestamp area city userid adid
    // 涵义： 时间戳     区域  城市  用户   广告

    // Application -> Kafka -> SparkStreaming -> Analysis

    var producer: KafkaProducer[String, String] = null
    if (switch) {
      val prop = new Properties()
      prop.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "linux1:9092")
      prop.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer")
      prop.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer")

      producer = new KafkaProducer[String, String](prop)
    }
    while (true) {
      mockData().foreach(
        data => {
          if (switch) {
            // 向kafka中生成数据
            val record = new ProducerRecord[String, String]("at", data)
            producer.send(record)
          }
          println(data)
        }
      )
      Thread.sleep(2000)
    }

  }

  def mockData() = {
    val list = new ListBuffer[String]()
    val areaList = ListBuffer[String]("华北", "华东", "华南")
    val cityList = ListBuffer[String]("北京", "上海", "深圳")
    for (i <- 1 to 30) {
      val area = areaList(new Random().nextInt(areaList.length))
      val city = cityList(new Random().nextInt(cityList.length))
      val userid = new Random().nextInt(6) + 1
      val adid = new Random().nextInt(6) + 1
      list.append(s"${System.currentTimeMillis()} ${area} ${city} ${userid} ${adid}")
    }
    list
  }

}
